package com.ddl.kafka.config.direct;

import com.ddl.kafka.config.MessageData;
//import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

import lombok.extern.slf4j.Slf4j;

/**
 * 消息生产者
 */
@Component
@Slf4j
public class CountProducer {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Value("${amqp.testExchange}")
    private String testExchange;

    @Value("${amqp.testRoutingKey}")
    private String testRoutingKey;

    @Value("${amqp.testQueueName}")
    private String testQueueName;

    public void syncSend(String msg) {
        // 创建 消息
        MessageData messageData = new MessageData();
        messageData.setMsg(msg);
        log.info("开始发送");
        // 同步发送消息
        rabbitTemplate.convertAndSend(testExchange, testRoutingKey, messageData);
    }

    @Async
    public ListenableFuture<Void> asyncSend(String msg) {
        try {
            // 发送消息
            this.syncSend(msg);
            // 返回成功的 Future
            return AsyncResult.forValue(null);
        } catch (Throwable ex) {
            // 返回异常的 Future
            return AsyncResult.forExecutionException(ex);
        }
    }

    /**
     * 批量发送消息
     */
    /*@Autowired
    private BatchingRabbitTemplate batchingRabbitTemplate;

    public void batchSyncSend(Integer id, String msg) {
        // 创建 Demo05Message 消息
        MessageData messageData = new MessageData();
        messageData.setMsg(msg);
        messageData.setId(id);
        // 同步发送消息
        batchingRabbitTemplate.convertAndSend(testExchange, testRoutingKey, messageData);
    }*/
}
